package com.rent.recommend

import com.mongodb.casbah.{MongoClient, MongoClientURI}
import com.mongodb.casbah.commons.MongoDBObject
import org.apache.log4j.Logger
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/*
@������
Happy,happy everyday!
��Ѽ��
2021/11/4

*/
case class MongoConfig(uri:String, db:String)
case class ESConfig(httpHosts:String, transportHosts:String, index:String, clustername:String)
case class House(hid: Int, houseSingleUrl: String, title: String, singleType: String, size: String, directType: String, layer: String, rentType: String, place: String, underPlace: String, schoolPlace: String , price: Double,imgUrl:String)
case class Hplace(pid: Int , name: String, lon: Double , lai: Double)

object DataLoader {

  private var log = Logger.getLogger("log");
  val HOUSE_PLACE_PATH = "D:\\IntelliJ IDEA-workspace\\HouseRent\\HouseRecommend\\NewDataLoader\\src\\main\\resources\\house.csv";
  val ES_HOUSE_INDEX = "house"
  val MONGODB_HOUSE_COLLECTION = "House"

  def main(args: Array[String]): Unit = {

    val config = Map(
      "spark.cores" -> "local[*]",
      "es.httpHosts" -> "localhost:9200",
      "es.transportHosts" -> "localhost:9300",
      "es.index" -> "house",
      "es.cluster.name" -> "elasticsearch",
      "mongo.uri" -> "mongodb://localhost:27017/houserec",
      "mongo.db" -> "houserec"
    )

    // ����һ��sparkConf
    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")

    // ����һ��SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    import spark.implicits._
    // ������һ�� ES ���õ���ʽ����
    // implicit val esConfig = ESConfig(config("es.httpHosts"), config("es.transportHosts"), config("es.index"), config("es.cluster.name"))
    //    val houseDF = spark.sql("select hid,")
    log.debug("---------------------------------------------------------------------------------------")
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
    val houseRDD = spark.sparkContext.textFile(HOUSE_PLACE_PATH)
    val houseDF = houseRDD.map(item => {
      val attr = item.split(",")
      log.debug(item)

      House(attr(0).toInt,
        attr(1).trim,
        attr(2).trim,
        attr(3).trim,
        attr(4).trim,
        attr(5).trim,
        attr(6).trim,
        attr(7).trim,
        attr(8).trim,
        attr(9).trim,
        attr(10).trim,
        attr(11).toDouble,
        attr(12).trim)

    }).toDF()

    // �����ݱ��浽MongoDB
    storeDataInMongoDB(houseDF)

    storeDataInES(houseDF)

    // �ر� Spark
    spark.stop()

  }

  def storeDataInES(movieDF:DataFrame): Unit = {
    movieDF
      .write
      .option("es.nodes","localhost:9200")
      .option("es.http.timeout","100m")
      .option("es.mapping.id","hid")
      .mode("overwrite")
      .format("org.elasticsearch.spark.sql")
      .save("house"+"/"+ES_HOUSE_INDEX)
  }

  def storeDataInMongoDB(browseDF: DataFrame)(implicit mongoConfig: MongoConfig): Unit = {
    // �½�һ��mongodb������
    val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))

    // ���mongodb���Ѿ�����Ӧ�����ݿ⣬��ɾ��
    mongoClient(mongoConfig.db)(MONGODB_HOUSE_COLLECTION).dropCollection()
    // ��DF����д���Ӧ��mongodb����
    browseDF.write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_HOUSE_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    //�����ݱ�����
    mongoClient(mongoConfig.db)(MONGODB_HOUSE_COLLECTION).createIndex(MongoDBObject("hid" -> 1))


    mongoClient.close()

  }
}
